package org.idea.mq.framework.rocketmq.provider;

import org.apache.rocketmq.client.MQAdmin;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.idea.mq.framework.rocketmq.common.FastJsonSerializer;
import org.idea.mq.framework.rocketmq.common.MqMsgSerializer;

import java.util.Scanner;

/**
 * @Author linhao
 * @Date created in 8:47 上午 2022/2/20
 */
public class ProviderApplication {

    public static final String GROUP_NAME = "mq-framework-application-provider";
    public static final String NAME_SERVER = "localhost:9876";
    public static final Integer SEND_TIME_OUT = 1000;
    public static final String USER_PAY_TOPIC = "user-pay-topic";
    public static final String ORDER_DELETE_TOPIC = "order-delete-topic";
    public static MqMsgSerializer mqMsgSerializer = new FastJsonSerializer();
    public static DefaultMQProducer defaultMQProducer;

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //底层会按照这个groupName将producer对象存储到一个Map集合中
        defaultMQProducer = new DefaultMQProducer(GROUP_NAME,false);
        defaultMQProducer.setNamesrvAddr(NAME_SERVER);
        defaultMQProducer.setSendMsgTimeout(SEND_TIME_OUT);
        defaultMQProducer.setVipChannelEnabled(false);
        //启动的核心方法在这里
        defaultMQProducer.start();
        System.out.println("==============请选择需要发送堆消息==============");
        while (true){
            Scanner scanner = new Scanner(System.in);
            String nextLine = scanner.nextLine();
            if("A".equals(nextLine)){
                sendTopicA();
            }
        }
    }

    public static void sendTopicA(){
        try {
            SendResult sendResult = defaultMQProducer.send(createMessage("pull-topic","*","push-topic"));
            System.out.println(sendResult);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void sendTopicB(){
        try {
            SendResult sendResult = defaultMQProducer.send(createMessage("topic-B","*","topic-B-msg"));
            System.out.println(sendResult);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private static Message createMessage(String topic, String tag, Object t) {
        return new Message(topic,tag,mqMsgSerializer.serializer(t));
    }
}
